//package service;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import org.apache.spark.SparkConf;
//import org.apache.spark.api.java.JavaPairRDD;
//import org.apache.spark.api.java.JavaRDD;
//import org.apache.spark.api.java.JavaSparkContext;
//import org.apache.spark.api.java.function.*;
//import scala.Tuple2;
//
//import java.io.Serializable;
//import java.util.Iterator;
//
//public class SparkRddAlarm implements Serializable {
//    public static void main(String[] args) {
//        SparkConf conf = new SparkConf().setAppName("SparkStremAlarmService").setMaster("local[7]");
//        JavaSparkContext sc = new JavaSparkContext(conf);
//
//        JavaRDD<String> stringJavaRDD = sc.textFile("C:\\Users\\user\\Desktop\\abc.txt");
//        JavaPairRDD<String, Integer> pairs = stringJavaRDD.mapToPair(line -> {
//            JSONObject jsonObject = JSON.parseObject(line);
//            String metricGroupId = jsonObject.getString("metricGroupId");
//            String metric = jsonObject.getString("metric");
//            Integer count = jsonObject.getInteger("count");
//            return new Tuple2<String, Integer>(metricGroupId + "_" + metric, count);
//        });
//        JavaPairRDD<String, Iterable<Integer>> group = pairs.groupByKey();
//        JavaPairRDD<String, Integer> top2score = group.mapToPair(tuple -> {
//            int i = 0;
//            Iterator<Integer> it = tuple._2.iterator();
//            while (it.hasNext()) {
//                i += it.next();
//            }
//            return new Tuple2<String, Integer>(tuple._1, i);
//        });
//
//        JavaPairRDD<String, Integer> res = top2score.reduceByKey(new Function2<Integer, Integer, Integer>() {
//            private static final long serialVersionUID = 1L;
//
//            @Override
//            public Integer call(Integer a, Integer b) throws Exception {
//                return a + b;
//            }
//        });
//
//        res.foreach(new VoidFunction<Tuple2<String, Integer>>() {
//            private static final long serialVersionUID = 1L;
//
//            @Override
//            public void call(Tuple2<String, Integer> t) throws Exception {
//                System.out.println(t._1 + " : " + t._2);
//            }
//        });
//    }
//}